package state;

import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class BroadCastDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        UserAction ac1 = new UserAction(1001l, "login");
        UserAction ac2 = new UserAction(1003l, "pay");
        UserAction ac3 = new UserAction(1002l, "car");
        UserAction ac4 = new UserAction(1001l, "logout");
        UserAction ac5 = new UserAction(1003l, "car");
        UserAction ac6 = new UserAction(1003l, "logout");
        DataStreamSource<UserAction> acs = env.fromElements(ac1, ac2, ac3, ac4, ac5, ac6);

        acs.filter(new FilterFunction<UserAction>() {
            @Override
            public boolean filter(UserAction userAction) throws Exception {
                return false;
            }
        }).setParallelism(8);
        MyPattern myPattern1 = new MyPattern("login", "logout");
        MyPattern myPattern2 = new MyPattern("car", "logout");
        DataStreamSource<MyPattern> patterns = env.fromElements(myPattern1);
        KeyedStream<UserAction, Long> keyed = acs.keyBy(value -> value.getUserId());

        MapStateDescriptor<Void, MyPattern> dec = new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(MyPattern.class));

        BroadcastStream<MyPattern> broadcast = patterns.broadcast(dec);


    }

    public static class PatternEvaluator extends KeyedBroadcastProcessFunction<Long, UserAction, MyPattern, Tuple2<Long, MyPattern>> {


        ValueState<String> prevActionsState;
        @Override
        public void open(Configuration parameters) throws Exception {
            prevActionsState = getRuntimeContext().getState(new ValueStateDescriptor<String>("lastAction", Types.STRING));
        }

        @Override
        public void processElement(UserAction value, ReadOnlyContext ctx, Collector<Tuple2<Long, MyPattern>> out) throws Exception {

            ReadOnlyBroadcastState<Void, MyPattern> patterns = ctx.getBroadcastState(new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(MyPattern.class)));
            MyPattern myPattern = patterns.get(null);
            String preAction = prevActionsState.value();
            if (myPattern != null && preAction != null) {
                if (myPattern.getFirstAction().equals(preAction) && myPattern.getSecondAction().equals(value.getUserAction())) {
                    out.collect(new Tuple2<>(ctx.getCurrentKey(), myPattern));
                }
            }
            prevActionsState.update(value.getUserAction());

        }

        @Override
        public void processBroadcastElement(MyPattern value, Context ctx, Collector<Tuple2<Long, MyPattern>> out) throws Exception {
            BroadcastState<Void, MyPattern> bcstate = ctx.getBroadcastState(new MapStateDescriptor<>("patterns", Types.VOID, Types.POJO(MyPattern.class)));
            bcstate.put(null, value);
        }
    }




}
